package com.sinozo.data.utils;

import com.sinozo.data.bean.TransientSink;
import com.sinozo.data.common.DataConfig;
import lombok.SneakyThrows;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import ru.yandex.clickhouse.ClickHousePreparedStatement;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class MyClickHouseUtil {

    public static <T> SinkFunction<T> getSinkFunction(String sql) {
        return JdbcSink.<T>sink(sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {

                        //使用反射的方式获取t对象中的数据
                        Class<?> tClz = t.getClass();

                        //获取并遍历属性
                        Field[] declaredFields = tClz.getDeclaredFields();
                        int offset = 0;
                        for (int i = 0; i < declaredFields.length; i++) {

                            //获取单个属性
                            Field field = declaredFields[i];
                            field.setAccessible(true);

                            //尝试获取字段上的自定义注解
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);
                            if (transientSink != null) {
                                offset++;
                                continue;
                            }

                            //获取属性值
                            Object value = field.get(t);

                            //给占位符赋值
                            preparedStatement.setObject(i + 1 - offset, value);
                        }
                        String sql = ((ClickHousePreparedStatement) preparedStatement).asSql();
                        System.out.println(sql + ";");
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchIntervalMs(3000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(DataConfig.CLICKHOUSE_DRIVER)
                        .withUrl(DataConfig.CLICKHOUSE_URL)
                        .withUsername(StringUtils.isEmpty(DataConfig.CLICKHOUSE_USER) ? null : DataConfig.CLICKHOUSE_USER)
                        .withPassword(StringUtils.isEmpty(DataConfig.CLICKHOUSE_PASSWORD) ? null : DataConfig.CLICKHOUSE_PASSWORD)
                        .build());
    }

//    public static <T>T executeSqlQuery(String sql) {
//        ClickHouseProperties properties = new ClickHouseProperties();
//        String url = DataConfig.CLICKHOUSE_URL;
//        properties.setSessionId("default-session-id");
//        properties.setDatabase(DataConfig.CLICKHOUSE_DATABASE);
////        properties.setHost("hadoop102");
//
//        ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
//        Map<ClickHouseQueryParam, String> additionalDBParams = new HashMap<>();
//        additionalDBParams.put(ClickHouseQueryParam.SESSION_ID, "new-session-id");
//
//        ClickHouseConnection conn = null;
//        try {
//            conn = dataSource.getConnection();
//            ClickHouseStatement stmt = conn.createStatement();
//            ResultSet rs = stmt.executeQuery(sql, additionalDBParams);
//
//            while (rs.next()) {
//                Long user_id = rs.getLong(1);
//                Long item_id = rs.getLong(2);
//                //System.out.println(user_id+" "+item_id);
//            }
//        } catch (SQLException e) {
//            throw new RuntimeException(e);
//
//        }
//
//    }

}
